#include <sys/epoll.h>
#include "miniutil/socketmul_epoll.h"

namespace miniutil { namespace multinet {

#ifndef WIN32

SocketMulplexService_Epoll::SocketMulplexService_Epoll()
{
    mepollfd = epoll_create(1024);
    std::cout<<"Create EPOLL FD OK"<<std::endl;
}

SocketMulplexService_Epoll::~SocketMulplexService_Epoll()
{
    close(mepollfd);
}
    
    
void SocketMulplexService_Epoll::run()
{
    struct epoll_event events[10000]; 
    stopstus = false;
    while(true)
    {
        if(stopcmd)
        {
            stopcmd = false;
            break;
        }
        
        do
        {
            miniutil::auto_lock kk(&msockmtx);
            initPollFd();
            //std::cout<<"begin epoll_wait"<<std::endl;
            int nfds = epoll_wait(mepollfd,  events, 10000, 1);
            if(nfds == -1)
            {
                std::cout<<"epoll_wait return -1"<<std::endl;
                perror(NULL);
                //epoll wait error
                thread::sleep(10);
                continue;
            }
            else if(nfds == 0)
            {
                //std::cout<<"epoll_wait return 0"<<std::endl;
                thread::sleep(10);
                continue;
            }
            
            //std::cout<<"epoll_wait return "<<nfds<<std::endl;
            time_t now;
            time(&now);
            if(now > mprechktime)
            {
                left_dquota = max_dquota;
                left_uquota = max_uquota;
                mprechktime = now;
            }
            
            for( int i=0; i< nfds; i++)
            {
                 if((events[i].events & EPOLLIN)  != 0x00 && left_dquota > 0)
                    do_readable(events[i].data.fd);
                 if((events[i].events & EPOLLOUT) != 0x00 && left_uquota > 0)
                    do_writable(events[i].data.fd);
                 if((events[i].events & EPOLLHUP) != 0x00|| (events[i].events & EPOLLERR) != 0x00)
                    do_exception(events[i].data.fd);
            }
        }while(0);
    }
    
}

void SocketMulplexService_Epoll::do_readable(int fd)
{
    do
    {
        if(mtcpchecksocs.find(fd) != mtcpchecksocs.end())
        {        
            MonTcpSocketInstance &ins = malltcpsockets[fd];
            if((ins.waitevt & EVT_INCOMING) != 0x00)
            {
                std::cout<<"[do_readable] some one coming!"<<std::endl;
                ins.socevent |= EVT_INCOMING;
            }
            else
            {
                std::pair<char*, unsigned int> writepos;
                ins.recvbuf.getwritepos(writepos);
                assert(writepos.second > 0);
                {
                    ins.recvret = ::recv(ins.soc, writepos.first,writepos.second, 0);
                    if(ins.recvret > 0)
                    {
                        ins.recvbuf.more_content(ins.recvret);
                        left_dquota -= ins.recvret;
                        if(left_dquota < 0)
                            break;
                    }
                    else
                    {
                        //std::cout<<"recv error"<<std::endl;
                    }
                }
                ins.socevent |= EVT_READ_READY;
           }
        }
        else if(mudpchecksocs.find(fd) != mudpchecksocs.end())
        {
                socklen_t nmln = sizeof(struct sockaddr);
                MonUdpSocketInstance &ins = malludpsockets[fd];
                MonUdpSocketInstance::UDPMSG* pmsg = new MonUdpSocketInstance::UDPMSG();
                pmsg->mlen = ::recvfrom(fd, pmsg->mpbuf, MonUdpSocketInstance::UDPMSG_SIZE, 0, (struct sockaddr*)&pmsg->maddr, &nmln);
                ins.vrecvbuf.push_back(pmsg);
        }
        else
        {
        //assert!
        }
    }while(0);
}

void SocketMulplexService_Epoll::do_writable(int fd)
{
    do
    {
        if(mtcpchecksocs.find(fd) != mtcpchecksocs.end())
        {
             MonTcpSocketInstance &ins = malltcpsockets[fd];
             if((ins.waitevt & EVT_CONNECTED) != 0x00)
                ins.socevent |= EVT_CONNECTED;
            else if(/*(ins.waitevt & EVT_WRITE_READY) != 0x00*/ ins.send_queue.size() > 0)
            {
                while ( ins.send_queue.size() > 0 )
                {
                    miniutil::data::shared_buffer b = ins.send_queue.front();
                    int sendlen =  (int)b->length() > left_uquota?left_uquota:(int)b->length();
                    if(sendlen <= 0)
                        break;

                    int len = ::send( fd, b->data(), (int)b->length(), 0 );
                    if ( len == (int)b->length() )
                    {
                        M_DEBUG("do_send( )len %d == length()%d\n", len, (int)b->length());
                        ins.send_queue.pop_front();
                        left_uquota -= len;
                        if(left_uquota <= 0)
                            break;
                        continue;
                    }
                    if ( len < 0)
                    {                        
                    /*
                        int ret = miniutil::socket::getLastError();
                        */
                        int ret = errno;
                        if(ret != EWOULDBLOCK)
                            ins.socevent |= EVT_WRITE_READY;
                        //M_DEBUG("send_len < 0 \n");
                    }
                    else if ( len > 0 && len <  (int)b->length() )
                    {
                        //M_DEBUG("do_send( )len %d<length()%d\n", len, (int)b->length());
                        b->erase( 0 , (size_t)len );
                        left_uquota -= len;
                    }
                    else if(len == 0)
                    {
                        ins.socevent |= EVT_WRITE_READY;
                    }
                    break;
                }
            }
       }
        else if(mudpchecksocs.find(fd) != mudpchecksocs.end())
        {
            socklen_t nmln = sizeof(struct sockaddr);
            MonUdpSocketInstance &ins = malludpsockets[fd];
            if(ins.vrecvbuf.size() > 0)
            {
                MonUdpSocketInstance::UDPMSG* pmsg = ins.vrecvbuf[0];
                ::sendto(ins.soc, pmsg->mpbuf, pmsg->mlen, 0, (struct sockaddr*)&pmsg->maddr, nmln);
                delete pmsg;
                ins.vrecvbuf.erase(ins.vrecvbuf.begin());
            }
      }
        else
        {
            //assert
        }
   }while(0);
}

void SocketMulplexService_Epoll::do_exception(int fd)
{
    if(mtcpchecksocs.find(fd) != mtcpchecksocs.end())
    {        
        MonTcpSocketInstance &ins = malltcpsockets[fd];
        ins.socevent |= EVT_EXCEPTION;
    }
}

void SocketMulplexService_Epoll::checkTcpAndRemove(int fd)
{
    bool found = false;
    if(mtcpchecksocs.find(fd) != mtcpchecksocs.end())
    {
        found = true;
    }
    
    if(found)
    {
        epoll_ctl(mepollfd, EPOLL_CTL_DEL, fd, NULL);
        mtcpchecksocs.erase(fd);
    }
 }

void SocketMulplexService_Epoll::checkUdpAndRemove(int fd)
{
    bool found = false;
    if(mudpchecksocs.find(fd) != mudpchecksocs.end())
    {
        found = true;
    }
    
    if(found)
    {
        epoll_ctl(mepollfd, EPOLL_CTL_DEL, fd, NULL);
        mudpchecksocs.erase(fd);
    }
}

void SocketMulplexService_Epoll::checkTcpAndUpdate(int fd, int event)
{
//??? the reason using mtcpchecksocs is merge EPOLL_CTL_ADD/EPOLL_CTL_MOD ???
    bool found = false;
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = event;
   if(mtcpchecksocs.find(fd) != mtcpchecksocs.end())
    {
        found = true;
    }    
    if(!found)
    {
        std::cout<<"[checkTcpAndUpdate]Add one Event!"<<std::endl;
        epoll_ctl(mepollfd, EPOLL_CTL_ADD, fd, &ev);
        std::cout<<"[checkTcpAndUpdate]insert to mtcpchecksocs!"<<std::endl;
        mtcpchecksocs.insert(fd);
    }
    else
    {
        epoll_ctl(mepollfd, EPOLL_CTL_MOD, fd, &ev);
    }
}

void SocketMulplexService_Epoll::checkUdpAndUpdate(int fd, int event)
{
    bool found = false;
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = event;
    if(mudpchecksocs.find(fd) != mudpchecksocs.end())
    {
        found = true;
    }    
    if(!found)
    {
        epoll_ctl(mepollfd, EPOLL_CTL_ADD, fd, &ev);
        mudpchecksocs.insert(fd);
    }
    else
    {
        epoll_ctl(mepollfd, EPOLL_CTL_MOD, fd, &ev);
    }
}

void SocketMulplexService_Epoll::initPollFd()
{
    std::map<SOCKET, MonTcpSocketInstance>::iterator iter_vsmt;
    for(iter_vsmt = malltcpsockets.begin(); iter_vsmt != malltcpsockets.end();)
    {
          if(iter_vsmt->second.closeflag)
          {
            checkTcpAndRemove(iter_vsmt->first);
            malltcpsockets.erase(iter_vsmt++);
            continue;
          }
        
        int waitevt = 0;
        
        MonTcpSocketInstance &ains = iter_vsmt->second;
        if((ains.waitevt & EVT_INCOMING ) != 0x00)
        {
            if((ains.socevent & EVT_INCOMING) == 0x00)
                waitevt = EPOLLIN;
        }
        else if((ains.waitevt & EVT_CONNECTED) != 0x00)
        {
            if((ains.socevent & EVT_CONNECTED) == 0x00)
                waitevt = EPOLLOUT;
        }
        else
        {
            if((ains.waitevt & EVT_READ_READY) != 0x00)
            {
                if(ains.recvret > 0 && ains.recvbuf.have_space(1024))
                     waitevt = EPOLLIN;                
            }
            if((ains.waitevt & EVT_WRITE_READY) != 0x00)
            {
                if(ains.sendret > 0 && ains.send_queue.size() > 0)
                    waitevt = EPOLLOUT;
            }
         }
         
        if(waitevt != 0)
        {   
            //std::cout<<"Update one Socket Event"<<std::endl;
            checkTcpAndUpdate(iter_vsmt->first, waitevt);
        }
        
        iter_vsmt ++;
    }
            
    std::map<SOCKET, MonUdpSocketInstance>::iterator iter_msmu;
    for(iter_msmu = malludpsockets.begin(); iter_msmu != malludpsockets.end();)
    {
        if(iter_msmu->second.closeflag)
        {
            checkUdpAndRemove(iter_msmu->first);
            malludpsockets.erase(iter_msmu++);
            continue;
        }
        //low eff
        int waitevt = 0;
        if(iter_msmu->second.vsendbuf.size() > 0)
        {
            waitevt |= EPOLLOUT;
        }
        waitevt |= EPOLLIN;
        checkUdpAndUpdate(iter_msmu->first, waitevt);
        iter_msmu ++;
    }            
}

#endif

} }
